package com.atguigu.transform;

import com.atguigu.func.ClickSource;
import com.atguigu.pojo.Event;
import com.atguigu.pojo.Wordcount;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author Mr.Zheng
 * @Date 2023/6/13 22:11
 *
 * 聚合算子：
 *  简单聚合：sum min minBy max maxBy
 *  规约聚合：reduce
 *
 *  Flink要求：要聚合，先keyBy
 *
 */
public class Flink02_AggTransform {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        //获取数据
        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        ds.print("input");

        //需求：统计每个人的点击次数
        ds.map(
                event -> new Wordcount(event.getUser(), 1)
        ).returns(Wordcount.class)
                .keyBy(
                        Wordcount::getWord
                )
                .sum("count");
                //.print();

        //需求：求每个user最后点击的数据

        ds.keyBy(
                Event::getUser
        )//.max("ts")
                .maxBy("ts")
                        .print();
        env.execute();

    }

}
